package com.zan.seckillservice.listener;

import com.zan.seckillservice.service.GoodsService;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;

/**
 * @Author Zan
 * @Create 2024/9/3 9:08
 * @ClassName: SeckillListener
 * @Description : MQ消费者
 */
@Component
@RocketMQMessageListener(
        topic = "seckillTopic4",
        consumerGroup = "seckill-consumer-group",
        consumeMode = ConsumeMode.CONCURRENTLY,
        consumeThreadNumber = 32
)
public class SeckillListener implements RocketMQListener<MessageExt> {

    @Resource
    private GoodsService goodsService;

    @Autowired
    private RedisTemplate redisTemplate;

    private final static int ZX_TIME = 10000; // 10s

    /**
     * 方案一：单机版（没法集群）
     * 扣减库存
     * 写订单表
     *
     * @param message
     */
    /*@Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody());
        // userId + "-" + goodsId
        Integer userId = Integer.parseInt(msg.split("-")[0]);
        Integer goodsId = Integer.parseInt(msg.split("-")[1]);
        // 扣减库存
        // jvm EntrySet waitSet
        synchronized (this) {
            // 这样子就确保事务先结束，再解锁，就可以保证数据的一致性，数据修改后，锁再解锁，不然的话事务都没结束，锁就结束了，别人又来了，导致一样的数据被减了两次，一模一样
            goodsService.realSeckill(userId, goodsId);
        }
    }*/


    // 方案二：如果是好几台机器，方案一是锁不住的，因此得用分布式锁
    // mysql（innodb行锁） or redis
    // 使用MySQL的行锁，但是不适合用于并发量特别大的场景（几百一两千还是可以的）
    /*@Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody());
        Integer userId = Integer.parseInt(msg.split("-")[0]);
        Integer goodsId = Integer.parseInt(msg.split("-")[1]);
        goodsService.realSeckill2(userId, goodsId);
    }*/

    // 方案三：redis，主要是通过setnx + 自旋，分布式锁，压力会分摊到redis和程序中，缓解db的压力
    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody());
        Integer userId = Integer.parseInt(msg.split("-")[0]);
        Integer goodsId = Integer.parseInt(msg.split("-")[1]);

        int currentThreadTime = 0;
        while (currentThreadTime < ZX_TIME) { // 如果ZX_TIME设定小了点，也会有些数据失败，因此可以设置大一点，true也行，但是不太好，因为有问题会一直循环，蛮占用资源的
            // 这里给一个过期时间，可以避免死锁的发生，如果使用true的时候，防止一直循环死锁
            Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock:" + goodsId, "", Duration.ofSeconds(30)); // 设置过期时间
            if (flag) {
                // 拿到锁成功
                try {
                    goodsService.realSeckill2(userId, goodsId);
                    return;
                } finally {
                    // 解锁 - 删除key
                    redisTemplate.delete("lock:" + goodsId);
                }
            } else {
                // 如何处理（拿不到锁）—— 自旋
                currentThreadTime += 200; // 自旋50次
                try {
                    Thread.sleep(200L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
